package com.ctl.juc;

import com.ctl.util.SnowflakeIdUtils;
import lombok.extern.slf4j.Slf4j;

import java.util.Queue;
import java.util.concurrent.*;

/**
 * <p>Title: ConcurrentLinkedQueueDemo</p>
 * <p>Description: </p>
 * <p>Copyright: Copyright (c) 2019</p>
 * <p>Company: www.ctl.com</p>
 *  在 Java 线程池中，线程可以处于多种不同状态，其中包括 TIMED_WAITING 和 WAITING。下面是这两种状态的详解：
 *     TIMED_WAITING 状态表示线程正在等待一个指定的时间段。这种状态通常发生在以下几种情况：
 *         调用了 Thread.sleep(long millis) 方法。
 *         调用了带超时参数的 Object.wait(long timeout) 方法。
 *         调用了带超时参数的 Thread.join(long millis) 方法。
 *         使用了带超时参数的 LockSupport.parkNanos(long nanos) 或 LockSupport.parkUntil(long deadline) 方法。
 *     WAITING 状态表示线程正在等待另一线程执行特定操作而进入无限期的等待状态。这种状态通常发生在以下几种情况：
 *         调用了没有超时参数的 Object.wait() 方法。
 *         调用了没有超时参数的 Thread.join() 方法。
 *         调用了 LockSupport.park() 方法。
 *
 * 1. LinkedBlockingQueue
 *   BlockingQueue<Runnable> queue = new LinkedBlockingQueue<>(100);
 *   LinkedBlockingQueue是一个基于链表实现的阻塞队列，通常用于生产者-消费者模型中。
 *   特点：
 *     支持容量限制，当队列满时进行阻塞。
 *     绘制了两个锁（一个用于入队，一个用于出队），以减少锁竞争。
 *     在高负载下具有较好的伸缩性。
 *   适用场景：
 *     需要一个有界队列，以防止过度消耗内存。
 *     任务生产和消费速率相对均衡的情况。
 * 2. ConcurrentLinkedQueue
 *   Queue<Runnable> queue = new ConcurrentLinkedQueue<>();
 *   ConcurrentLinkedQueue是一个基于链表实现的无界非阻塞队列，非常适合于高并发场景下的快速访问。
 *   特点：
 *     无界队列，不会阻塞生产者或消费者。
 *     使用乐观锁（CAS操作）来实现高效的并发访问。
 *     适用于高吞吐量的环境，但没有容量控制，因此需注意内存消耗。
 *   适用场景：
 *     无需容量限制，无需阻塞控制。
 *     高并发环境下需要快速非阻塞的队列操作。
 * 3. ArrayBlockingQueue
 *   BlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(100);
 *   ArrayBlockingQueue是基于数组实现的有界阻塞队列，提供固定大小的队列容量。
 *   特点：
 *     固定容量，防止过度消耗内存。
 *     单锁实现，可能在高并发情况下存在一定的锁竞争。
 *     顺序存储，元素访问速度较快。
 *   适用场景：
 *     需要容量限制。
 *     队列大小已知且不需动态调整。
 * 选择队列的参考依据
 *   容量限制：
 *     如果需要限制队列的大小以防止资源耗尽，应选择LinkedBlockingQueue或ArrayBlockingQueue。
 *     如果不需要容量限制，可以使用ConcurrentLinkedQueue。
 *   阻塞行为：
 *     如果希望在队列满时阻塞生产者或在队列空时阻塞消费者，应该选择LinkedBlockingQueue或ArrayBlockingQueue。
 *     如果不需要阻塞行为，并且更关注高并发性能，可以使用ConcurrentLinkedQueue。
 *   性能和并发要求：
 *     对于高并发场景，希望最大限度地减少锁争用，ConcurrentLinkedQueue是一个很好的选择。
 *     对于读取和写入比较均衡的情况，LinkedBlockingQueue表现优良，因为它分离了读写锁。
 *     如果追求更快的基本操作速度（如插入和删除），而且队列大小是固定的，可以考虑ArrayBlockingQueue。
 *   内存使用：
 *     LinkedBlockingQueue和ConcurrentLinkedQueue是基于链表的，适合动态增长的队列。
 *     ArrayBlockingQueue是基于数组的，适合固定大小的队列，并且在内存利用率上比较紧凑。
 * @author ctl
 * @version 1.1
 * @date 2024-06-27 10:48
 */
import java.util.concurrent.atomic.AtomicInteger;
 

@Slf4j
public class ConcurrentLinkedQueueDemo {
 
    static void pt(Queue<String> queue) {
        try {
            while (!Thread.currentThread().isInterrupted()) {
                int size = queue.size();
                if (size >= 1000) {
                    for (int i = 0; i < size; i++) {
                        String poll = queue.poll();
                        if (poll != null) {
                            log.info("msg={}", poll);
                        } else {
                            break;
                        }
                    }
                } else {
                    Thread.sleep(100);
                }
            }
            log.info("isInterrupted={}", Thread.currentThread().isInterrupted());
        } catch (Exception e) {
            log.error("pt.error", e);
        }
    }

    public static void main(String[] args) throws InterruptedException {

        Queue<String> queue = new ConcurrentLinkedQueue();
        ThreadPoolExecutors.threadPoolExecutor1.execute(() -> {
            for (int i = 0; i <= 1000000000; i++) {
                int finalI = i;
                ThreadPoolExecutors.threadPoolExecutor.execute(() -> {
                    queue.add("t1-" + SnowflakeIdUtils.genID() + "-" + finalI);
                });
                ThreadPoolExecutors.threadPoolExecutor.execute(() -> {
                    queue.add("t2-" + SnowflakeIdUtils.genID() + "-" + finalI);
                });
                ThreadPoolExecutors.threadPoolExecutor.execute(() -> {
                    queue.add("t3-" + SnowflakeIdUtils.genID() + "-" + finalI);
                });
            }
        });
        ThreadPoolExecutors.threadPoolExecutor1.execute(() -> {
            pt(queue);
        });
        ThreadPoolExecutors.threadPoolExecutor1.execute(() -> {
            pt(queue);
        });
        TimeUnit.MILLISECONDS.sleep(3000);
        ThreadPoolExecutors.threadPoolExecutor.shutdown();//方法用于停止接收新的任务，同时不影响已经提交的任务的执行
        ThreadPoolExecutors.threadPoolExecutor1.shutdown();//方法用于停止接收新的任务，同时不影响已经提交的任务的执行
        boolean b = ThreadPoolExecutors.threadPoolExecutor.awaitTermination(1, TimeUnit.SECONDS);
        log.info("awaitTermination={}", b);
        boolean b1 = ThreadPoolExecutors.threadPoolExecutor1.awaitTermination(1, TimeUnit.SECONDS);
        log.info("awaitTermination1={}", b1);
        if (!b1) {
            ThreadPoolExecutors.threadPoolExecutor1.shutdownNow();
            boolean b11 = ThreadPoolExecutors.threadPoolExecutor1.awaitTermination(1, TimeUnit.SECONDS);
            log.info("awaitTermination1-1={}", b11);
        }

        log.info("isTerminated={},isTerminated1={}", ThreadPoolExecutors.threadPoolExecutor.isTerminated(), ThreadPoolExecutors.threadPoolExecutor1.isTerminated());
    }
}
